package com.atguigu.windows;

import com.atguigu.bean.WaterSensor;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

import java.time.Duration;

public class Flink13_Time_EventTime02 {
    public static void main(String[] args) {
        Configuration configuration = new Configuration();
        configuration.setInteger("rest.port",10000);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(configuration);
        env.setParallelism(2);


        DataStreamSource<String> ds = env.socketTextStream("hadoop162", 9999);
        SingleOutputStreamOperator<String> mainStream = ds.map(new MapFunction<String, WaterSensor>() {
            @Override
            public WaterSensor map(String value) throws Exception {
                String[] datas = value.split(",");
                return new WaterSensor(datas[0], Long.valueOf(datas[1]), Integer.valueOf(datas[2]));
            }
        })
                .assignTimestampsAndWatermarks(WatermarkStrategy
                        //水印最大容忍迟到时间
                        .<WaterSensor>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                        //返回水印比较时间戳
                        .withTimestampAssigner(new SerializableTimestampAssigner<WaterSensor>() {
                            @Override
                            public long extractTimestamp(WaterSensor element, long recordTimestamp) {
                                return element.getTs();
                            }
                        })
                )
                .keyBy(new KeySelector<WaterSensor, String>() {
                    @Override
                    public String getKey(WaterSensor value) throws Exception {
                        return value.getId();
                    }
                })
                .process(new KeyedProcessFunction<String, WaterSensor, String>() {
                    private long timer;
                    //需求：监控水位传感器的水位值，如果水位值在5s之内(event time)连续上升，则报警。
                    //创建临时变量来判断是否是第一条数据进场
                    boolean isFirst = true;
                    Integer maxVc = 0;
                    //触发定时器的时候需要完成的操作！
                    @Override
                    public void onTimer(long timestamp,
                                        OnTimerContext ctx,
                                        Collector<String> out) throws Exception {

                        out.collect("水位连续5s上升了，红色预警！");
                        //这次预警之后，为下一次预警做准备，重新生成定时器
                        isFirst = true;
                    }

                    @Override
                    public void processElement(WaterSensor value,
                                               Context ctx,
                                               Collector<String> out) throws Exception {

                        //判断是否第一条数据进场
                        if (isFirst) {
                            //第一条数据进场
                            isFirst = false;
                            //启动定时器
                            //这里的时间后面也要用，所以上升到全局变量
                            System.out.println(value.getTs() + ",注册一个定时器！");
                            timer = value.getTs() + 5000;
                            ctx.timerService().registerEventTimeTimer(timer);

                            //顺便将当前的状态值保存起来，以供后面数据进场进行比较
                            maxVc = value.getVc();
                        } else {
                            //这是后面的数据进场了
                            //对水位值进行比较
                            if (maxVc > value.getVc()) {
                                //那么水位下降了，就可以取消定时器了
                                System.out.println("水位下降，取消定时器！！" + value.getVc());
                                ctx.timerService().deleteEventTimeTimer(timer);
                                //取消了定时器之后，又要马上重新启动一个新的定时器
                                timer = value.getTs() + 5000;
                                ctx.timerService().registerEventTimeTimer(timer);
                            }
                        }

                        //将水位值变换为最新状态下的水位
                        maxVc = value.getVc();


                    }
                });
        mainStream.print();

        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
